#include "media_stream.h"

#include <algorithm>
#include <inttypes.h>
#include <string.h>
#include <sys/prctl.h>
#include <sys/time.h>
#include <vector>
#include <cassert>
#include <memory>
#include <utility>
#include "utils_log.h"
#include "media_video_manager.h"
#include "media_audio_manager.h"

namespace El {
namespace Media {
StreamSourcePtr StreamSource::Create(int32_t ch, int32_t type)
{
    return std::make_shared<StreamSource>(ch, type);
}

StreamSource::StreamSource(int32_t ch, int32_t type) : ch_(ch), type_(type) {}

StreamSource::~StreamSource()
{
    Stop();
}

bool StreamSource::SetInfo(int32_t max, std::string name)
{
    max_ = max;
    name_ = name;
    return true;
}

int32_t StreamSource::Start()
{
    if (start_) {
        EL_WARN("StreamSource is already started ch:{}, type:{}", ch_, type_);
        return -1;
    }

    EL_INFO("StreamSource starting ch:{}, type:{}", ch_, type_);
    start_ = true;
    waitForIFrame_ = true;

    // 默认10帧缓存
    queue_ = std::make_unique<Utils::BlockingQueue<Media::MediaFramePtr>>(name_, max_);

    if (!callbackManager_.IsEmpty()) {
        streamThread_ = std::make_unique<std::thread>(&StreamSource::StreamThread, this);
        EL_INFO("StreamSource thread is starting ch:{}, type:{}", ch_, type_);
    }

    streamHandle_ = VideoStreamManager::GetInstance(ch_, type_).Start([this](const MediaFramePtr &frame) {
        if (waitForIFrame_) {
            if (frame->GetFrameType() != MEDIA_FRAME_I) {
                EL_DEBUG("Waiting for I-frame, dropping frame type:{}", frame->GetFrameType());
                return;
            }
            waitForIFrame_ = false;
            EL_INFO("Got first I-frame, starting normal operation");
        }

        bool ret = queue_->Push(frame, 0);
        if (!ret) {
            EL_ERROR("queue is full size:{}, drop frame {}, waiting for next I-frame", queue_->Size(),
                      frame->ToString().c_str());
            waitForIFrame_ = true;
        }
    });

    streamAudioHandle_ = MediaAudioManager::GetInstance(0).Start([this](const MediaFramePtr &frame) {
        if (waitForIFrame_) {
            return;
        }

        bool ret = queue_->Push(frame, 0);
        if (!ret) {
            EL_ERROR("queue is full size:{}, drop frame {} ", queue_->Size(), frame->ToString().c_str());
        }
    });

    // 立即请求I帧，确保快速显示首帧
    ForceIframe();
    return true;
}

void StreamSource::Stop()
{
    if (!start_) {
        return;
    }
    start_ = false;
    waitForIFrame_ = true;

    VideoStreamManager::GetInstance(ch_, type_).Stop(streamHandle_);
    streamHandle_ = 0;

    MediaAudioManager::GetInstance(0).Stop(streamAudioHandle_);
    streamAudioHandle_ = 0;

    queue_->SetActive(false);
    callbackManager_.Clear();
    if (streamThread_ && streamThread_->joinable()) {
        streamThread_->join();
        streamThread_.reset();
    }
    EL_INFO("StreamSource stop ch:{}, type:{}", ch_, type_);
}

int32_t StreamSource::Register(const Proc &proc)
{
    return callbackManager_.Register(proc);
}

void StreamSource::Unregister(int32_t handle)
{
    callbackManager_.Unregister(handle);
}

MediaFramePtr StreamSource::GetFrame(int32_t timeout)
{
    return queue_->Pop(timeout);
}

void StreamSource::StreamThread()
{
    while (start_) {
        auto media = queue_->Pop(20);
        if (!media) {
            continue;
        }
        callbackManager_.InvokeAll(media);
    }
    EL_INFO("StreamSource thread is stopping ch:{}, type:{}", ch_, type_);
}

bool StreamSource::ForceIframe()
{
    VideoStreamManager::GetInstance(ch_, type_).ForceIframe();
    return true;
}

} // namespace Media
} // namespace El